package com.heima.search.listener;

import com.alibaba.fastjson.JSON;
import com.heima.model.common.vo.SearchArticleVo;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component("SearchArticleListener")
public class ArticleListener {
    private static final String ES_TABLE_NAME = "app_info_article";
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    @KafkaListener(topics = "insert-article-topic")
    public void insertOrUpdate(String message) {
        //检验参数
        if(message!=null){
            try {
                //反序列对象
                SearchArticleVo searchArticleVo = JSON.parseObject(message, SearchArticleVo.class);
                //查询索引库中是否有存在该文章信息，如果存在则为修改，如果不存在，则添加
                SearchRequest searchRequest = new SearchRequest(ES_TABLE_NAME);
                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                searchSourceBuilder.query(QueryBuilders.matchQuery("id", searchArticleVo.getId()));
                searchRequest.source(searchSourceBuilder);
                //获取响应数据
                SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                //解析响应数据，并获得索引库是否有该文章信息
                long hitTotal = response.getHits().getTotalHits().value;
                if(hitTotal==0){
                    //向索引库添加文章信息
                    IndexArticleSource(ES_TABLE_NAME,searchArticleVo);
                    return ;
                }
                //修改索引库中文章信息(删除原来的，在创建新的文档)
                DeleteArticleSource(ES_TABLE_NAME,searchArticleVo.getId().toString());
                IndexArticleSource(ES_TABLE_NAME,searchArticleVo);
                return;
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    @KafkaListener(topics = "delete-article-topic")
    public void delete(String message) {
        //检验参数
        if(message!=null){
            //反序列化参数
            try {
                String articleId = JSON.parseObject(message, String.class);
                DeleteArticleSource(ES_TABLE_NAME, articleId);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
    private void IndexArticleSource(String ESTable, SearchArticleVo searchArticleVo) throws IOException {
        //向索引库添加文章信息
        IndexRequest source = new IndexRequest(ESTable).id(searchArticleVo.getId().toString()).source(JSON.toJSONString(searchArticleVo), XContentType.JSON);
        restHighLevelClient.index(source, RequestOptions.DEFAULT);
    }
    private void DeleteArticleSource(String ESTable,String id) throws IOException {
        DeleteRequest deleteRequest = new DeleteRequest(ESTable);
        deleteRequest.id(id);
        restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
    }
}
